package com.github.zhijieqing.socket.cluster.common.manager;


import com.github.zhijieqing.log.utils.JsonUtil;
import com.github.zhijieqing.socket.cluster.common.channel.MsgChannel;
import com.github.zhijieqing.socket.cluster.common.config.SocketClusterConfig;
import com.github.zhijieqing.socket.cluster.common.domain.DefaultSocketMsg;
import com.github.zhijieqing.socket.cluster.common.domain.SocketMsg;
import com.github.zhijieqing.socket.cluster.common.domain.TopicSocketMsg;
import com.github.zhijieqing.socket.cluster.common.listener.MessageListener;
import com.github.zhijieqing.socket.cluster.common.listener.PubSubListener;
import com.github.zhijieqing.socket.cluster.common.node.SocketServerNode;
import com.github.zhijieqing.socket.cluster.common.storage.ClientStorage;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import javax.annotation.PreDestroy;
import javax.websocket.Session;
import java.util.ArrayList;
import java.util.Collection;
import java.util.List;
import java.util.Set;

@Slf4j
public abstract class AbstractSocketManager implements SocketManager {

    @Getter
    @Setter
    private MsgChannel msgChannel;

    @Getter
    @Setter
    private ClientStorage clientStorage;

    @Getter
    @Setter
    private SocketClusterConfig socketClusterConfig;

    @Getter
    @Setter
    private List<MessageListener> messageListenerList = new ArrayList<>();

    @Override
    public void init() {
        getMsgChannel().init(this);
        getClientStorage().put(getCurrentServerNode());
    }

    @PreDestroy
    public void preDestroy() {
        getClientStorage().remove(getCurrentServerNode());
    }

    @Override
    public void put(String clientId, String sessionId) {
        getClientStorage().put(clientId, sessionId, getCurrentServerNode());
    }

    @Override
    public void put(String clientId, Session session) {
        getClientStorage().put(clientId, session, getCurrentServerNode());
    }

    @Override
    public Collection<SocketServerNode> getServerNodeByClientId(String clientId) {
        return getClientStorage().get(clientId);
    }

    @Override
    public List<Session> getWsSession(String clientId) {
        /*TODO*/
        return null;
    }

    @Override
    public void removeSession(String sessionId) {
        Set<String> clientIdSet = getClientStorage().getSessionClientCache().get(sessionId);
        if (clientIdSet == null) {
            return;
        }
        clientIdSet.forEach((clientId) -> {
            try {
                Set<String> sessionIdSet = getClientStorage().getClientSessionCache().get(clientId);
                sessionIdSet.remove(sessionId);
                if (sessionIdSet.size() == 0) {
                    removeCurrentNode(clientId);
                }
            } catch (Exception e) {
                log.error("Ws session[{}]关闭移除client[{}]移除", sessionId, clientId, e);
            }
        });
    }

    @Override
    public void removeAll(String clientId) {
        getClientStorage().remove(clientId);
    }

    @Override
    public void removeCurrentNode(String clientId) {
        getClientStorage().remove(clientId, getCurrentServerNode());
    }

    @Override
    public void remove(String clientId, Session session) {
        getClientStorage().remove(clientId, session, getCurrentServerNode());
    }

    @Override
    public void sendMessage(SocketMsg msg) {
        getMsgChannel().sendMsgToApp(getCurrentServerNode(), msg);
    }

    public void sendMessage(String clientId, String topic, Object payload) {
        TopicSocketMsg socketMsg = new TopicSocketMsg();
        socketMsg.setClientId(clientId);
        socketMsg.setTopic(topic);
        socketMsg.setPayload(payload);
        sendMessage(clientId, socketMsg);
    }

    public void sendMessage(String clientId, Object payload) {
        DefaultSocketMsg socketMsg = new DefaultSocketMsg();
        socketMsg.setClientId(clientId);
        socketMsg.setPayload(payload);
        sendMessage(clientId, socketMsg);
    }

    /**
     * 当前节点处理则不进行广播
     *
     * @param clientId
     * @param msg
     */
    public void sendMessageCurrentFirst(String clientId, SocketMsg msg) {
        if (getClientStorage().currentNodeContains(clientId, getCurrentServerNode())) {
            if (log.isDebugEnabled()) {
                log.debug("消息在本节点处理：{}", clientId);
            }
            onMessage(msg);
        } else {
            if (log.isDebugEnabled()) {
                log.debug("消息不在本节点处理，转发至其他节点：{}", clientId);
            }
            Collection<SocketServerNode> socketServerNodes = getServerNodeByClientId(clientId);
            socketServerNodes.forEach((socketServerNode -> {
                if (!getCurrentServerNode().getId().equals(socketServerNode.getId())) {
                    getMsgChannel().sendMsgToNode(socketServerNode, msg);
                }
            }));
        }
    }

    /**
     * 所有clientId相关节点都需处理
     *
     * @param clientId
     * @param msg
     */
    @Override
    public void sendMessage(String clientId, SocketMsg msg) {
        if (StringUtils.isEmpty(clientId)) {
            log.debug("clientId is null");
            return;
        }
        Collection<SocketServerNode> socketServerNodes = getServerNodeByClientId(clientId);
        socketServerNodes.forEach((socketServerNode -> {
            if (getCurrentServerNode().getId().equals(socketServerNode.getId())) {
                onMessage(msg);
            } else {
                getMsgChannel().sendMsgToNode(socketServerNode, msg);
            }
        }));
    }

    public void onMessage(SocketMsg msg) {
        /*TODO 可改用rxjava的发布订阅模式，或者guava观察者模式*/
        if (log.isDebugEnabled()) {
            log.debug("消息在本节点处理：{}", JsonUtil.writeValueAsString(msg));
        }
        messageListenerList.forEach((messageListener) -> {
            messageListener.onMessage(msg);
        });
    }

    public <M> void addListener(Class<M> type, MessageListener<M> listener) {
        getMessageListenerList().add(new PubSubListener(type, listener));
    }

    @Override
    public SocketServerNode getCurrentServerNode() {
        return getSocketClusterConfig().getServerNode();
    }

}

